import org.apache.flink.api.common.functions.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Arrays;

public class JavaWordCount {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString("taskmanager.numberOfTaskSolts","8");
        conf.setString("taskmanager.network.netty.sendReceiveBufferSize 4M","1024*1024*4");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);
        //1.读取数据 2.转换算子 3.输出
        //env.setParallelism(1);
        //env.disableOperatorChaining();
        DataStreamSource<String> ds = env.socketTextStream("nn2.hadoop", 6667);
/**
 *  1.lambda版本 比较麻烦
 */
//        ds.flatMap((String line, Collector<String> out)->{
//            String[] arr = line.split(" ");
//            //Arrays.asList(arr).forEach();
//            Arrays.stream(arr).forEach(t->out.collect(t));
//        }).returns(Types.STRING)
//                .map(t-> Tuple2.of(t,1)).returns(Types.TUPLE(Types.STRING,Types.INT))
//                //java中keyBy f0表第一位
//                .keyBy(t->t.f0).sum(1).print();
//
//        env.execute();

/**
 *  2.Function版本
 */
//        ds.flatMap(new FlatMapFunction<String, String>() {
//            @Override
//            public void flatMap(String line, Collector<String> out) throws Exception {
//                String[] arr = line.split(" ");
//                Arrays.stream(arr).forEach(f->out.collect(f));
//            }
//        }).map(new MapFunction<String, Tuple2<String,Integer>>() {
//            @Override
//            public Tuple2<String, Integer> map(String value) throws Exception {
//                return Tuple2.of(value,1);
//            }
//        }).keyBy(new KeySelector<Tuple2<String, Integer>, String>() {//有3个参数 一个函数 一个int0 一个string f0
//            @Override
//            public String getKey(Tuple2<String, Integer> stringIntegerTuple2) throws Exception {
//                return stringIntegerTuple2.f0;
//            }
//        }).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
//            @Override
//            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> t1, Tuple2<String, Integer> t2) throws Exception {
//                return Tuple2.of(t1.f0,t1.f1+t2.f1);
//            }
//        }).print();

        /**
         *  3.混合版本
         */
//        ds.flatMap(new FlatMapFunction<String, Tuple2<String,Integer>>() {
//            @Override
//            public void flatMap(String line, Collector<Tuple2<String, Integer>> out) throws Exception {
//                String[] arr = line.split(" ");
//                Arrays.stream(arr).forEach(f->out.collect(Tuple2.of(f,1)));
//            }
//        }).keyBy(0).sum(1).print();


        /**
         *  4.RichFunction写法
         */
//        ds.flatMap(new RichFlatMapFunction<String, String>() {
//            int indexOfThisSubtask=-1;
//            @Override
//            public void open(Configuration parameters) throws Exception {
//                indexOfThisSubtask = getRuntimeContext().getIndexOfThisSubtask();
//                System.out.println("innt method called once at beginning...");
//            }
//
//            @Override
//            public void close() throws Exception {
//                System.out.println("innt method called once at end...");
//            }
//
//            @Override
//            public void flatMap(String line, Collector<String> out) throws Exception {
//                System.out.println(indexOfThisSubtask+"----->"+line);
//                String[] arr = line.split(" ");
//                Arrays.stream(arr).forEach(f->out.collect(f));
//
//            }
//        }).map(new RichMapFunction<String, Tuple2<String,Integer>>() {
//            @Override
//            public Tuple2<String, Integer> map(String value) throws Exception {
//                return Tuple2.of(value,1);
//            }
//        }).keyBy(0).sum(1).print();

        /**
         *  5.最底层的processfunction
         */

        ds.setBufferTimeout(5000).process(new ProcessFunction<String, String>() {
            int subtask=-1;
            @Override
            public void open(Configuration parameters) throws Exception {
                subtask=getRuntimeContext().getIndexOfThisSubtask();
                super.open(parameters);
            }

            @Override
            public void close() throws Exception {
                super.close();
            }

            @Override
            public void processElement(String value, Context context, Collector<String> out) throws Exception {
                System.out.println(subtask+"--->"+value);
                String[] arr = value.split(" ");
                Arrays.stream(arr).forEach(f->out.collect(f));
            }
        }).setParallelism(2).process(new ProcessFunction<String, Tuple2<String,Integer>>() {
            @Override
            public void processElement(String word, Context context, Collector<Tuple2<String, Integer>> out) throws Exception {
                out.collect(Tuple2.of(word,1));
            }
        }).setParallelism(2).keyBy(0)
                .process(new KeyedProcessFunction<Tuple, Tuple2<String, Integer>, Tuple2<String, Integer>>() {//Tuple是key
            int sum = 0;
            @Override
            public void processElement(Tuple2<String, Integer> value, Context context, Collector<Tuple2<String, Integer>> out) throws Exception {
                sum+=value.f1;
                //out.collect(Tuple2.of(value.f0,sum));
                out.collect(Tuple2.of(context.getCurrentKey().toString(),sum));

            }
        }).setParallelism(2).print();
        System.out.println(env.getExecutionPlan());

        env.execute();
        

    }
}
